package com.czl.myRabbitMq.mqServer;

import com.czl.myRabbitMq.commom.BinaryTransformTool;
import com.czl.myRabbitMq.commom.Consume;
import com.czl.myRabbitMq.commom.Request;
import com.czl.myRabbitMq.commom.Response;
import com.czl.myRabbitMq.commom.argument.*;
import com.czl.myRabbitMq.commom.exception.mqException;
import com.czl.myRabbitMq.commom.returns.BasicReturns;
import com.czl.myRabbitMq.commom.returns.DescribeReturns;
import com.czl.myRabbitMq.mqServer.core.BasicProperties;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//TODO 后续扩展为多个支持多个Exchange的功能
@Slf4j
public class BrokerServer {
    private VirtualHost virtualHost = new VirtualHost("default");

    //获取网络连接
    private ServerSocket serverSocket = null;
    //这个map用来存储连接的数量 每有一个连接 就将该连接存入socketMap中
    //String 为channelID Socket为每一次的连接
    private final ConcurrentHashMap<String, Socket> socketMap = new ConcurrentHashMap<>();
    //线程池用来处理服务器连接过程中的连接
    private ExecutorService executorService = null;
    //控制服务器的开启与关闭
    private volatile boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }
    //服务器启动的方法
    public void start() throws IOException {
        log.info("[BrokerServer]服务器开始启动");
        executorService = Executors.newCachedThreadPool();
        while (runnable){
            Socket client = serverSocket.accept();
            //获取到连接过后直接将处理连接的过程丢给线程池
            executorService.submit(()->{
               processConnection(client);
            });
        }
    }
    //停止服务器 一般停止进程即可 这里设置一个方法 可以方便后续进行单元测试
    public void stop() throws IOException {
        runnable = false;
        //让线程池连接全部关闭 同时放弃当前执行的任务
        executorService.shutdownNow();
        serverSocket.close();
    }

    private void processConnection(Socket client) {
        //先获取到一个文件的输入输出流
        try(InputStream inputStream = client.getInputStream();
            OutputStream outputStream = client.getOutputStream()){
            //因为我们需要按照一定格式来对数据进行一个读取
            try(DataInputStream dataInputStream = new DataInputStream(inputStream);
            DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                //此时来进行数据的读取 因为dataInputStream在数据读取结束之后会抛出异常来终止循环
                while (true){
                    //读取请求并解析
                    Request request = readRequest(dataInputStream);
                    //根据请求计算响应
                    Response response = process(request, client);
                    //将响应写回到dataOutputStream对象
                    writeResponse(response, dataOutputStream);
                }
            }catch (EOFException | SocketException e){
                //dataInputStream在读取结束后会抛出异常，同时客户端主动断开连接也会抛出异常
                log.info("连接断开" + client.getInetAddress());
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            //最后需要关闭连接
            try{
                client.close();
                clearSession(client);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private void clearSession(Socket client) {
        //遍历上述键值对然后删除
        //每次处理完请求之后需要将client相关的键值对全部删除

        //在集合遍历过程中不能边遍历边删除 会破坏跌倒器的结构 所以我们将需要删除的内容存入集合中
        List<String> deleteDatas = new ArrayList<>();
        for (Map.Entry<String, Socket> map: socketMap.entrySet()) {
            if(map.getValue() == client){
                deleteDatas.add(map.getKey());
            }
        }

        for(String socketKey : deleteDatas){
            socketMap.remove(socketKey);
        }
        log.info("成功清楚会话连接 + " + deleteDatas.toString() );
    }

    //写回请求的内容
    private void writeResponse(Response response, DataOutputStream dataOutputStream) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getBody());
        //写入消息记得刷新
        dataOutputStream.flush();
    }

    //处理中间消息的过程 这里我们规定
    //0x1 创建channel 0x2销毁channel
    //0x3 创建Exchange 0x4销毁Exchange
    //0x5 创建Queue 0x6销毁Queue
    //0x7 创建Binding 0x8销毁Binding
    //0x9 发送Message 0xa 订阅Message
    //0xb 返回ACK

    //       0xc服务器给客户端推送消息
    private Response process(Request request, Socket client) throws mqException {
        //进行参数的一个初步解析
        //因为所有Argument中的类都是继承了BasicArguments 所以我们可以将该类初步解析为BasicArguments
        BasicArguments basicArguments = (BasicArguments) BinaryTransformTool.fromData(request.getBody());
        log.info("ChannelId = #{}",basicArguments.getChannelId());
        //判断这个类的类型
        switch (request.getType()){
            // 创建一个channel
            case 0x1: {
                String channelId = basicArguments.getChannelId();
                socketMap.put(channelId, client);
                log.info("创建channel成功ChannelId = " + channelId);
                break;
            }
            //销毁一个channel
            case 0x2: {
                socketMap.remove(basicArguments.getChannelId());
                log.info("销毁channel成功ChannelId = " + basicArguments.getChannelId());
                break;
            }
            //创建一个Exchange
            case 0x3: {
                ExchangeDeclareArguments exchangeArguments = (ExchangeDeclareArguments) basicArguments;
                virtualHost.exchangeDeclare(exchangeArguments.getExchangeName(), exchangeArguments.getExchangeType(),
                        exchangeArguments.isDurable(), exchangeArguments.isAutoDelete(), exchangeArguments.getParams());
                break;
            }
            //删除一个交换机
            case 0x4: {
                DeleteExchangeArguments deleteExchangeArguments =(DeleteExchangeArguments) basicArguments;
                virtualHost.deleteExchange(deleteExchangeArguments.getExchangeName());
                break;
            }
            //创建一个队列
            case 0x5: {
                QueueDeclareArguments queueDeclareArguments = (QueueDeclareArguments) basicArguments;
                virtualHost.queueDeclare(queueDeclareArguments.getQueueName(), queueDeclareArguments.isDurable(),
                        queueDeclareArguments.isExclusive(), queueDeclareArguments.isAutoDelete(),
                        queueDeclareArguments.getParams());
                break;
            }
            //销毁队列
            case 0x6: {
                DeleteQueueArguments deleteQueueArguments = (DeleteQueueArguments) basicArguments;
                virtualHost.deleteQueue(deleteQueueArguments.getQueueName());
                break;
            }
            //创建一个绑定
            case 0x7: {
                QueueBindingArguments queueBindingArguments = (QueueBindingArguments) basicArguments;
                virtualHost.queueBinding(queueBindingArguments.getExchangeName(), queueBindingArguments.getQueueName(),
                        queueBindingArguments.getBindingKey());
                break;
            }
            //删除一个绑定
            case 0x8: {
                QueueUnbindingArguments queueUnbindingArguments = (QueueUnbindingArguments) basicArguments;
                virtualHost.queueUnbinding(queueUnbindingArguments.getExchangeName(),
                        queueUnbindingArguments.getQueueName());
                break;
            }
            //发送一个Message
            case 0x9: {
                BasicPublishArguments basicPublishArguments = (BasicPublishArguments) basicArguments;
                virtualHost.basicPublish(basicPublishArguments.getExchangeName(), basicPublishArguments.getRoutineKey(),
                        basicPublishArguments.getExchangeType(), basicPublishArguments.getBasicProperties(),
                        basicPublishArguments.getBody());
                break;
            }
            //订阅消息
            case 0xa: {
                BasicConsumeArguments basicConsumeArguments = (BasicConsumeArguments) basicArguments;
                virtualHost.basicConsume(basicConsumeArguments.getConsumeTag(),
                        basicConsumeArguments.getQueueName(), basicConsumeArguments.isAutoACK(),
                        new Consume() {
                            //这里的服务器中的回调函数都是执行同一个逻辑 找到订阅当前队列的客户端
                            //然后源源不断地向客户端发送消息
                            @Override
                            public void delivery(String consumeTag, BasicProperties basicProperties, byte[] body) throws IOException {
                                //这里的consumeTag就是请求中对应的channelId 这么先使用consumeTag 来找出对应的连接
                                Socket client = socketMap.get(consumeTag);

                                //构造出响应 body部分
                                DescribeReturns describeReturns = new DescribeReturns();
                                describeReturns.setConsumerTag(consumeTag);
                                describeReturns.setBasicProperties(basicProperties);
                                describeReturns.setBody(body);
                                //rid本质是对应请求和响应的
                                //但这里订阅客户端后 无需请求就会一直收到消息 所以后续rid可以无需设置
                                describeReturns.setRid("");
                                describeReturns.setOk(true);
                                describeReturns.setChannelID(consumeTag);
                                byte[] binaryData = BinaryTransformTool.getBinaryData(describeReturns);

                                //构造响应数据
                                Response response = new Response();
                                request.setType(0xc);
                                request.setLength(binaryData.length);
                                request.setBody(binaryData);

                                //将响应写回到连接的输出流中
                                //同时这个流对象不能关闭 一旦关闭client.getOutputStream()的对象也会关闭
                                //后续将无法进行消息的写入了
                                DataOutputStream dataOutputStream = new DataOutputStream(client.getOutputStream());
                                writeResponse(response, dataOutputStream);
                            }
                        });
                break;
            }
            case 0xb: {
                BasicACKArguments basicACKArguments = (BasicACKArguments) basicArguments;
                virtualHost.basicACK(basicACKArguments.getQueueName(), basicACKArguments.getMessageId());
                break;
            }
            default:{
                throw new mqException("不存在响应的响应格式");
            }
        }

        //最后进行响应数据的构造
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setOk(true);
        basicReturns.setChannelID(basicArguments.getChannelId());
        basicReturns.setRid(basicReturns.getRid());
        //将其序列化
        byte[] data = BinaryTransformTool.getBinaryData(basicReturns);
        Response response = new Response();
        response.setType(0xc);
        response.setLength(data.length);
        response.setBody(data);
        log.info("消息请求处理成功");
        return response;
    }


    //读取请求的内容
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        byte[] body = new byte[request.getLength()];
        int end = dataInputStream.read(body);
        if(end != body.length){
            throw new IOException("文件读取异常");
        }
        request.setBody(body);
        return request;
    }
}
